package main

import (
	"flag"
	"runtime"
	"time"

	"github.com/influxdata/influxdb/client/v2"
	log "github.com/sirupsen/logrus"
	"k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/clientcmd"
	"strings"
)

func main() {
	log.Info("begin")
	ipaddr := flag.String("ip", "", "fluxdb ip")
	port := flag.String("port", "8086", "port")
	userName := flag.String("user", "", "username")
	password := flag.String("password", "", "password")
	database := flag.String("database", "", "database")
	var kubeconfig *string
	switch runtime.GOOS {
	case "darwin":
		kubeconfig = flag.String("config", "./config", "absolute path to the kubeconfig file")
	case "windows":
		kubeconfig = flag.String("config", "D:\\Go\\gopath\\src\\k8s.io\\kubernetes\\panda\\middle\\cluster\\event_watch\\config", "absolute path to the kubeconfig file")
	case "linux":
		kubeconfig = flag.String("config", "/root/.kube/config", "absolute path to the kubeconfig file")
	}
	flag.Parse()

	config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
	if err != nil {
		log.Errorf("build config from Flag err :%s", err)
	}
	log.Info(*kubeconfig)
	return
	if *ipaddr == "" {
		log.Errorf("influxdb can't be null")
		return
	} else if *database == "" {
		log.Errorf("database can't be null")
	}

	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		log.Errorf("get clientset error %s", err)
		return
	}

	cli, err := getInfluxDBClient(*ipaddr, *port, *userName, *password)
	if err != nil {
		log.Errorf("influxdb connect failure: %s", err)
		return
	}

	watchlist := cache.NewListWatchFromClient(clientset.CoreV1().RESTClient(), "events", v1.NamespaceAll,
		fields.Everything())
	_, controller := cache.NewInformer(
		watchlist,
		&v1.Event{},
		time.Second*0,
		cache.ResourceEventHandlerFuncs{
			AddFunc: func(obj interface{}) {
				event := obj.(*v1.Event)
				log.Infof("resource name: %s, add", event.Name)
				eventInsertToInfluxdb(event, cli, *database)
			},
			DeleteFunc: func(obj interface{}) {
				event := obj.(*v1.Event)
				log.Infof("resource name: %s, delete", event.Name)
			},
			UpdateFunc: func(oldObj, newObj interface{}) {
				event := newObj.(*v1.Event)
				log.Infof("resource name: %s, update ", event.Name)
				eventInsertToInfluxdb(event, cli, *database)
			},
		},
	)
	stop := make(chan struct{})
	go controller.Run(stop)
	for {
		time.Sleep(time.Second)
	}
	// 获取influxdb 客户端
	func getInfluxDBClient(ip, port, username, password string) (client.Client, error) {
		addr := ip + ":" + port
		log.Info(addr, username, password)
		client, err := client.NewHTTPClient(client.HTTPConfig{
			Addr:     addr,
			Username: "",
			Password: "",
		})
		return client, err
	}

}

// 将event时间存入influxdb
func eventInsertToInfluxdb(event *v1.Event, cli client.Client, database string) {

	//select * from marathon_log order by time desc limit 100;
	if event.Count > 1 && event.InvolvedObject.Kind == "Pod" && (event.Reason == "Unhealthy" || event.Reason == "Started") {
		appNameList := strings.Split(event.Name, "-")
		appId := strings.Join(appNameList[0:len(appNameList)-2], "-")
		tags := map[string]string{"app_id": appId, "event_type": event.Reason}
		fields := map[string]interface{}{
			"cnt_id":   strings.Split(event.Name, ".")[0],
			"evt_desc": event.Message,
			"evt_time": event.LastTimestamp.Format("01-02 15:04:05"),
			"ip_addr":  event.Source.Host,
		}

		_, err := writeData(cli, database, "marathon_log", tags, fields, time.Now())
		if err != nil {
			log.Errorf("write data to influxd err: %s", err)
		}
	}
}

// WriteData write data
func writeData(cli client.Client, dataBase string, measurement string, tags map[string]string, fields map[string]interface{}, t time.Time) (bool, error) {
	bp, err := client.NewBatchPoints(client.BatchPointsConfig{
		Database: dataBase,
	})
	if err != nil {
		return false, err
	}
	// Create a point and add to batch
	point, err := client.NewPoint(measurement, tags, fields, t)
	if err != nil {
		return false, err
	}
	bp.AddPoint(point)

	// Write the batch
	if err := cli.Write(bp); err != nil {
		return false, err
	}
	return true, nil
}
